package com.yifeng.repo.base.mdm.transport.worker;

import com.talkyun.utils.Looper;
import com.yifeng.repo.base.api.DateTimeRange;
import com.yifeng.repo.base.constant.BaseConstant;
import com.yifeng.repo.base.mdm.MasterDataManager;
import com.yifeng.repo.base.mdm.transport.biz.BizTransportAction;
import com.yifeng.repo.base.utils.converter.DateHelper;
import com.yifeng.repo.base.utils.redis.worker.RedisLockWorker;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.Collections;
import java.util.List;
import java.util.Map;

/**
 * 周期性数据传输：支持拉取增量
 * Created by daibing on 2024/2/6.
 */
@Slf4j
public class DataTransportPeriodicWorker<T> {
    private static final DateTimeFormatter TIDY_DATE_FORMATTER = DateTimeFormatter.ofPattern(DateHelper.TIDY_DAY_FORMAT);
    private static final Integer THREE_DAY_EXPIRE_SECONDS = 3 * 24 * 60 * 60;
    private static final Integer LOOPER_INTERVAL_MILLIS = 60 * 1000;
    private final String redisKeyPrefix;
    private final RedisLockWorker redisLockWorker;
    private final MasterDataManager masterDataManager;
    private final BizTransportAction<T> bizTransportAction;
    private Looper worker;

    public DataTransportPeriodicWorker(String redisKeyPrefix, RedisLockWorker redisLockWorker,
                                       MasterDataManager masterDataManager, BizTransportAction<T> bizTransportAction) {
        this.redisKeyPrefix = redisKeyPrefix;
        this.redisLockWorker = redisLockWorker;
        this.masterDataManager = masterDataManager;
        this.bizTransportAction = bizTransportAction;
    }

    public void init() {
        worker = buildWorker();
        worker.start();
    }

    public void destroy() {
        worker.shutdown();
    }

    /**
     * 1分钟运行一次, 拉取一次周期内增量数据
     */
    private Looper buildWorker() {
        // 根据拉取数据间隔，计算每天多少周期，比如拉取时间间隔15分钟一次，每天96个时间段 (0 ~ 95)
        int totalPeriodByDay = 24 * 60 / bizTransportAction.fetchPeriodMinutes();
        String name = bizTransportAction.name() + "-periodic";
        return new Looper(name, LOOPER_INTERVAL_MILLIS, 1000) {
            @Override
            protected void loop() throws Throwable {
                // 1. 按时间段切分统计
                String today = LocalDate.now().format(TIDY_DATE_FORMATTER);
                String periodKey = redisKeyPrefix.concat(BaseConstant.REDIS_SEPARATOR).concat(name)
                        .concat(BaseConstant.REDIS_SEPARATOR).concat(today);
                long periodValue = redisLockWorker.incr(periodKey, totalPeriodByDay - 1, THREE_DAY_EXPIRE_SECONDS);
                String segmentKey = today + String.format("%04d", periodValue);

                // 2. 结束时间小于当前时间，当前时间段数据还没录入完毕，睡眠等待
                DateTimeRange dateTimeRange = new DateTimeRange(
                        LocalTime.MIN.plusMinutes(periodValue * bizTransportAction.fetchPeriodMinutes()).atDate(LocalDate.now()),
                        bizTransportAction.fetchPeriodMinutes(),
                        true
                );
                while (LocalDateTime.now().isBefore(dateTimeRange.getEnd())) {
                    doSleep(LOOPER_INTERVAL_MILLIS);
                }

                // 3. 拉取数据，并且投递给消费者
                int count = 0;
                Map<String, Object> offsetMap = Collections.emptyMap();
                do {
                    List<T> dataList = bizTransportAction.fetchData(segmentKey, dateTimeRange, offsetMap, masterDataManager);
                    if (dataList.isEmpty()) {
                        break;
                    }

                    count += dataList.size();
                    offsetMap = bizTransportAction.saveData(segmentKey, dataList);
                } while (true);

                // 4. 拉取完成
                bizTransportAction.finished(segmentKey, dateTimeRange, count);
            }

            @Override
            protected void loopThrowable(Throwable t) {
                log.warn("PeriodicWorker failed: name={} error=", name, t);
            }
        };
    }

}
